/*
 * Copyright (c) 2021 The red-star Project
 *
 * Licensed under the Apache License, version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.inyourcode.core.transport.session.iml;

import com.inyourcode.core.serialization.api.Serializer;
import com.inyourcode.core.serialization.api.SerializerFactory;
import com.inyourcode.core.serialization.api.SerializerType;
import com.inyourcode.core.transport.api.payload.BytesHolder;
import com.inyourcode.core.transport.api.payload.JRequestBytes;
import com.inyourcode.core.transport.api.payload.JResponseBytes;
import com.inyourcode.core.transport.session.HandlerWrapper;
import com.inyourcode.core.transport.session.RequestContext;
import com.inyourcode.core.transport.session.ResponseContext;
import com.inyourcode.core.transport.session.SessionHandlerMapping;
import com.inyourcode.core.transport.session.api.IMessageCoder;
import com.inyourcode.core.transport.session.api.MessageHolder;
import com.inyourcode.core.transport.session.head.ProtocolHeadMsgPack;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author JackLei
 */
public class MessageCoder4MessagePack implements IMessageCoder {
    private static final Logger LOGGER = LoggerFactory.getLogger(MessageCoder4MessagePack.class);

    @Override
    public byte serializerType() {
        return SerializerType.MSGPACK.value();
    }

    @Override
    public BytesHolder encode(MessageHolder messageHolder) {
        byte serializerType = SerializerType.MSGPACK.value();
        Serializer serializer = SerializerFactory.getSerializer(serializerType);
        int traceId = messageHolder.getTraceId();
        long invokeId = messageHolder.getInvokeId();

        Object message = messageHolder.getData();
        byte[] messageBytes = serializer.writeObject(message);

        ProtocolHeadMsgPack headBuilder = new ProtocolHeadMsgPack();
        headBuilder.setTraceId(traceId);
        headBuilder.setMessage(messageBytes);
        headBuilder.setTimeStamp(messageHolder.getTimeStamp());

        if (messageHolder instanceof RequestContext) {
            JRequestBytes jRequestPayload = new JRequestBytes(invokeId);
            jRequestPayload.bytes(serializerType, serializer.writeObject(headBuilder));
            return jRequestPayload;
        } else if (messageHolder instanceof ResponseContext) {
            JResponseBytes jResponseBytes = new JResponseBytes(invokeId);
            jResponseBytes.bytes(serializerType, serializer.writeObject(headBuilder));
            return jResponseBytes;
        } else {
            return null;
        }
    }

    @Override
    public MessageHolder decode(BytesHolder bytesHolder) {
        byte serializerCode = bytesHolder.serializerCode();
        long invokeId = bytesHolder.invokeId();
        Serializer serializer = SerializerFactory.getSerializer(serializerCode);
        byte[] bytes = bytesHolder.bytes();

        if (serializer == null) {
            LOGGER.error("The serializer could not be found when type:{}", serializerCode);
            return null;
        }

        HandlerWrapper wrapper = SessionHandlerMapping.getHandlerWrapper(invokeId);
        if (wrapper == null) {
            LOGGER.error("The wrapper could not be found when invokeId: {}", invokeId);
            return null;
        }

        ProtocolHeadMsgPack head = serializer.readObject(bytes, ProtocolHeadMsgPack.class);
        if (head == null) {
            LOGGER.error("The protocol head is null when invokeId: {}", invokeId);
            return null;
        }

        byte[] messageBytes = head.getMessage();
        Object message = serializer.readObject(messageBytes, wrapper.getMessageClazz());

        if (bytesHolder instanceof JRequestBytes) {
            RequestContext requestContext = new RequestContext(invokeId, head.getTraceId(), message, serializerCode, head.getTimeStamp());
            return requestContext;
        } else  if (bytesHolder instanceof JResponseBytes) {
            ResponseContext responseContext = new ResponseContext(invokeId, head.getTraceId(), message, serializerCode, head.getTimeStamp());
            return responseContext;
        } else {
            return null;
        }

    }

}
